---
id: core-application
title: Core application - Python SDK
sidebar_label: Core application
description: Learn to develop and customize Workflows and Activities using the Temporal Python SDK, manage parameters, set timeouts, execute Activities, and run a Worker Process efficiently.
toc_max_heading_level: 2
keywords:
  - temporal python core
  - develop temporal workflow
  - temporal python activities
  - workflow logic requirements
  - customizing workflow type
  - activity parameters
  - activity return values
  - activity execution in workflow
  - activity timeouts
  - getting activity results
  - temporal python worker
  - registering workflow types
tags:
  - Activities
  - Temporal Client
  - Task Queues
  - Workers
  - Workflows
  - Python SDK
  - Temporal SDKs
---

This page shows how to do the following:

- [Develop a basic Workflow](#develop-workflows)
- [Define Workflow parameters](#workflow-parameters)
- [Define Workflow return parameters](#workflow-return-values)
- [Customize your Workflow Type](#workflow-type)
- [Develop Workflow logic](#workflow-logic-requirements)
- [Develop a basic Activity](#develop-activities)
- [Develop Activity Parameters](#activity-parameters)
- [Define Activity return values](#activity-return-values)
- [Customize your Activity Type](#activity-type)
- [Start an Activity Execution](#activity-execution)
- [Set the required Activity Timeouts](#required-timeout)
- [Get the results of an Activity Execution](#get-activity-results)
- [Run a Worker Process](#run-a-dev-worker)
- [Register types](#register-types)

## Develop a basic Workflow {#develop-workflows}

**How to develop a basic Workflow using the Temporal Python SDK.**

Workflows are the fundamental unit of a Temporal Application, and it all starts with the development of a [Workflow Definition](/workflows#workflow-definition).

In the Temporal Python SDK programming model, Workflows are defined as classes.

Specify the `@workflow.defn` decorator on the Workflow class to identify a Workflow.

Use the `@workflow.run` to mark the entry point method to be invoked. This must be set on one asynchronous method defined on the same class as `@workflow.defn`. Run methods have positional parameters.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_workflows_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )
```

### Define Workflow parameters {#workflow-parameters}

**How to define Workflow parameters using the Temporal Python SDK.**

Temporal Workflows may have any number of custom parameters.
However, we strongly recommend that objects are used as parameters, so that the object's individual fields may be altered without breaking the signature of the Workflow.
All Workflow Definition parameters must be serializable.

Workflow parameters are the method parameters of the singular method decorated with `@workflow.run`.
These can be any data type Temporal can convert, including [`dataclasses`](https://docs.python.org/3/library/dataclasses.html) when properly type-annotated.
Technically this can be multiple parameters, but Temporal strongly encourages a single `dataclass` parameter containing all input fields.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_dataobject_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from dataclasses import dataclass
# ...
# ...
@dataclass
class YourParams:
    greeting: str
    name: str
```

### Define Workflow return parameters {#workflow-return-values}

**How to define Workflow return parameters using the Temporal Python SDK.**

Workflow return values must also be serializable.
Returning results, returning errors, or throwing exceptions is fairly idiomatic in each language that is supported.
However, Temporal APIs that must be used to get the result of a Workflow Execution will only ever receive one of either the result or the error.

To return a value of the Workflow, use `return` to return an object.

To return the results of a Workflow Execution, use either `start_workflow()` or `execute_workflow()` asynchronous methods.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_workflows_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )
```

### Customize your Workflow Type {#workflow-type}

**How to customize your Workflow Type using the Temporal Python SDK.**

Workflows have a Type that are referred to as the Workflow name.

The following examples demonstrate how to set a custom name for your Workflow Type.

You can customize the Workflow name with a custom name in the decorator argument. For example, `@workflow.defn(name="your-workflow-name")`. If the name parameter is not specified, the Workflow name defaults to the unqualified class name.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_workflows_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )
```

### Develop Workflow logic {#workflow-logic-requirements}

**How to develop Workflow logic using the Temporal Python SDK.**

Workflow logic is constrained by [deterministic execution requirements](/workflows#deterministic-constraints).
Therefore, each language is limited to the use of certain idiomatic techniques.
However, each Temporal SDK provides a set of APIs that can be used inside your Workflow to interact with external (to the Workflow) application code.

Workflow code must be deterministic. This means:

- no threading
- no randomness
- no external calls to processes
- no network I/O
- no global state mutation
- no system date or time

All API safe for Workflows used in the [`temporalio.workflow`](https://python.temporal.io/temporalio.workflow.html) must run in the implicit [`asyncio` event loop](https://docs.python.org/3/library/asyncio-eventloop.html) and be _deterministic_.

## Develop a basic Activity {#develop-activities}

**How to develop a basic Activity using the Temporal Python SDK.**

One of the primary things that Workflows do is orchestrate the execution of Activities.
An Activity is a normal function or method execution that's intended to execute a single, well-defined action (either short or long-running), such as querying a database, calling a third-party API, or transcoding a media file.
An Activity can interact with world outside the Temporal Platform or use a Temporal Client to interact with a Temporal Service.
For the Workflow to be able to execute the Activity, we must define the [Activity Definition](/activities#activity-definition).

You can develop an Activity Definition by using the `@activity.defn` decorator.
Register the function as an Activity with a custom name through a decorator argument, for example `@activity.defn(name="your_activity")`.

:::note

The Temporal Python SDK supports multiple ways of implementing an Activity:

- Asynchronously using [`asyncio`](https://docs.python.org/3/library/asyncio.html)
- Synchronously multithreaded using [`concurrent.futures.ThreadPoolExecutor`](https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor)
- Synchronously multiprocess using [`concurrent.futures.ProcessPoolExecutor`](https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor) and [`multiprocessing.managers.SyncManager`](https://docs.python.org/3/library/multiprocessing.html#multiprocessing.managers.SyncManager)

Blocking the async event loop in Python would turn your asynchronous program into a synchronous program that executes serially, defeating the entire purpose of using `asyncio`.
This can also lead to potential deadlock, and unpredictable behavior that causes tasks to be unable to execute.
Debugging these issues can be difficult and time consuming, as locating the source of the blocking call might not always be immediately obvious.

Due to this, consider not make blocking calls from within an asynchronous Activity, or use an async safe library to perform
these actions.
If you must use a blocking library, consider using a synchronous Activity instead.

:::

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_activities_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from temporalio import activity
# ...
# ...
@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"
```

### Develop Activity Parameters {#activity-parameters}

**How to develop Activity Parameters using the Temporal Python SDK.**

There is no explicit limit to the total number of parameters that an [Activity Definition](/activities#activity-definition) may support.
However, there is a limit to the total size of the data that ends up encoded into a gRPC message Payload.

A single argument is limited to a maximum size of 2 MB.
And the total size of a gRPC message, which includes all the arguments, is limited to a maximum of 4 MB.

Also, keep in mind that all Payload data is recorded in the [Workflow Execution Event History](/workflows#event-history) and large Event Histories can affect Worker performance.
This is because the entire Event History could be transferred to a Worker Process with a [Workflow Task](/workers#workflow-task).

{/* TODO link to gRPC limit section when available */}

Some SDKs require that you pass context objects, others do not.
When it comes to your application data—that is, data that is serialized and encoded into a Payload—we recommend that you use a single object as an argument that wraps the application data passed to Activities.
This is so that you can change what data is passed to the Activity without breaking a function or method signature.

Activity parameters are the function parameters of the function decorated with `@activity.defn`.
These can be any data type Temporal can convert, including dataclasses when properly type-annotated.
Technically this can be multiple parameters, but Temporal strongly encourages a single dataclass parameter containing all input fields.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_activities_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from temporalio import activity
from your_dataobject_dacx import YourParams

# ...
# ...
@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"
```

### Define Activity return values {#activity-return-values}

**How to define Activity return values using the Temporal Python SDK.**

All data returned from an Activity must be serializable.

There is no explicit limit to the amount of data that can be returned by an Activity, but keep in mind that all return values are recorded in a [Workflow Execution Event History](/workflows#event-history).

An Activity Execution can return inputs and other Activity values.

The following example defines an Activity that takes a string as input and returns a string.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_activities_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
# ...
@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"
```

### Customize your Activity Type {#activity-type}

**How to customize your Activity Type**

Activities have a Type that are referred to as the Activity name.
The following examples demonstrate how to set a custom name for your Activity Type.

You can customize the Activity name with a custom name in the decorator argument. For example, `@activity.defn(name="your-activity")`.
If the name parameter is not specified, the Activity name defaults to the function name.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_activities_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
# ...
@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"
```

## Start an Activity Execution {#activity-execution}

**How to start an Activity Execution using the Temporal Python SDK.**

Calls to spawn [Activity Executions](/activities#activity-execution) are written within a [Workflow Definition](/workflows#workflow-definition).
The call to spawn an Activity Execution generates the [ScheduleActivityTask](/references/commands#scheduleactivitytask) Command.
This results in the set of three [Activity Task](/workers#activity-task) related Events ([ActivityTaskScheduled](/references/events#activitytaskscheduled), [ActivityTaskStarted](/references/events#activitytaskstarted), and ActivityTask[Closed])in your Workflow Execution Event History.

A single instance of the Activities implementation is shared across multiple simultaneous Activity invocations.
Activity implementation code should be _idempotent_.

The values passed to Activities through invocation parameters or returned through a result value are recorded in the Execution history.
The entire Execution history is transferred from the Temporal service to Workflow Workers when a Workflow state needs to recover.
A large Execution history can thus adversely impact the performance of your Workflow.

Therefore, be mindful of the amount of data you transfer through Activity invocation parameters or Return Values.
Otherwise, no additional limitations exist on Activity implementations.

To spawn an Activity Execution, use the [`execute_activity()`](https://python.temporal.io/temporalio.workflow.html#execute_activity) operation from within your Workflow Definition.

`execute_activity()` is a shortcut for [`start_activity()`](https://python.temporal.io/temporalio.workflow.html#start_activity) that waits on its result.

To get just the handle to wait and cancel separately, use `start_activity()`.
In most cases, use `execute_activity()` unless advanced task capabilities are needed.

A single argument to the Activity is positional. Multiple arguments are not supported in the type-safe form of `start_activity()` or `execute_activity()` and must be supplied by the `args` keyword argument.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_workflows_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )
```

### Set the required Activity Timeouts {#required-timeout}

**How to set the required Activity Timeouts using the Temporal Python SDK.**

Activity Execution semantics rely on several parameters.
The only required value that needs to be set is either a [Schedule-To-Close Timeout](/encyclopedia/detecting-activity-failures#schedule-to-close-timeout) or a [Start-To-Close Timeout](/encyclopedia/detecting-activity-failures#start-to-close-timeout).
These values are set in the Activity Options.

Activity options are set as keyword arguments after the Activity arguments.

Available timeouts are:

- schedule_to_close_timeout
- schedule_to_start_timeout
- start_to_close_timeout

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/activity_timeouts_retires/your_workflows_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
# ...
        activity_timeout_result = await workflow.execute_activity(
            your_activity,
            YourParams(greeting, "Activity Timeout option"),
            # Activity Execution Timeout
            start_to_close_timeout=timedelta(seconds=10),
            # schedule_to_start_timeout=timedelta(seconds=10),
            # schedule_to_close_timeout=timedelta(seconds=10),
        )
```

### Get the results of an Activity Execution {#get-activity-results}

**How to get the results of an Activity Execution using the Temporal Python SDK.**

The call to spawn an [Activity Execution](/activities#activity-execution) generates the [ScheduleActivityTask](/references/commands#scheduleactivitytask) Command and provides the Workflow with an Awaitable.
Workflow Executions can either block progress until the result is available through the Awaitable or continue progressing, making use of the result when it becomes available.

Use [`start_activity()`](https://python.temporal.io/temporalio.workflow.html#start_activity) to start an Activity and return its handle, [`ActivityHandle`](https://python.temporal.io/temporalio.workflow.ActivityHandle.html). Use [`execute_activity()`](https://python.temporal.io/temporalio.workflow.html#execute_activity) to return the results.

You must provide either `schedule_to_close_timeout` or `start_to_close_timeout`.

`execute_activity()` is a shortcut for `await start_activity()`. An asynchronous `execute_activity()` helper is provided which takes the same arguments as `start_activity()` and `await`s on the result. `execute_activity()` should be used in most cases unless advanced task capabilities are needed.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/your_workflows_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from temporalio import workflow
# ...
# ...
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )
```

## Run a Worker Processes {#run-a-dev-worker}

**How to run a Worker Process using the Temporal Python SDK.**

The [Worker Process](/workers#worker-process) is where Workflow Functions and Activity Functions are executed.

- Each [Worker Entity](/workers#worker-entity) in the Worker Process must register the exact Workflow Types and Activity Types it may execute.
- Each Worker Entity must also associate itself with exactly one [Task Queue](/workers#task-queue).
- Each Worker Entity polling the same Task Queue must be registered with the same Workflow Types and Activity Types.

A [Worker Entity](/workers#worker-entity) is the component within a Worker Process that listens to a specific Task Queue.

Although multiple Worker Entities can be in a single Worker Process, a single Worker Entity Worker Process may be perfectly sufficient.
For more information, see the [Worker tuning guide](/develop/worker-performance).

A Worker Entity contains a Workflow Worker and/or an Activity Worker, which makes progress on Workflow Executions and Activity Executions, respectively.

To develop a Worker, use the `Worker()` constructor and add your Client, Task Queue, Workflows, and Activities as arguments.
The following code example creates a Worker that polls for tasks from the Task Queue and executes the Workflow.
When a Worker is created, it accepts a list of Workflows in the workflows parameter, a list of Activities in the activities parameter, or both.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/run_worker_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
from temporalio.client import Client
from temporalio.worker import Worker
# ...
# ...
async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="your-task-queue",
        workflows=[YourWorkflow],
        activities=[your_activity],
    )
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())
```

### Register types {#register-types}

**How to register types using the Temporal Python SDK.**

All Workers listening to the same Task Queue name must be registered to handle the exact same Workflows Types and Activity Types.

If a Worker polls a Task for a Workflow Type or Activity Type it does not know about, it fails that Task.
However, the failure of the Task does not cause the associated Workflow Execution to fail.

When a `Worker` is created, it accepts a list of Workflows in the `workflows` parameter, a list of Activities in the `activities` parameter, or both.

<div className="copycode-notice-container">
  <a href="https://github.com/temporalio/documentation/blob/main/sample-apps/python/your_app/run_worker_dacx.py">
    View the source code
  </a>{' '}
  in the context of the rest of the application code.
</div>

```python
# ...
async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="your-task-queue",
        workflows=[YourWorkflow],
        activities=[your_activity],
    )
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())
```
